/**
 * FileName: MqttConfig
 * Author: Wb
 * Date: 2025/9/24 10:56
 * Description:
 * History:
 */
package org.test.device.config.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.StringUtils;

/**
 * 〈一句话功能简述〉<br>
 * 〈mqtt 配置〉
 *
 * @author Wb
 * @create 2025/9/24
 * @since 1.0.0
 */
@Slf4j
@Configuration
public class MqttConfig {

    // 从配置读取：broker、主题、客户端前缀、用户名/密码
    @Value("${mqtt.broker:tcp://localhost:1883}")
    private String mqttServer;

    @Value("${mqtt.subscribeTopic:device}")
    private String subscribeTopic;

    @Value("${mqtt.clientIdPrefix:Test-Devices}")
    private String clientIdPrefix;

    @Value("${mqtt.username:}")
    private String username;

    @Value("${mqtt.password:}")
    private String password;

    /**
     * 创建连接工厂
     * @return
     */
    @Bean
    public MqttPahoClientFactory mqttPahoClientFactory(){
        DefaultMqttPahoClientFactory factory=new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();

        //配置MQTT服务地址
        options.setServerURIs(new String[]{mqttServer});
        log.info("MQTT服务器地址配置: {}", mqttServer);

        //认证：若配置了用户名则密码认证，否则匿名
        if (StringUtils.hasText(username)) {
            options.setUserName(username);
            options.setPassword(password != null ? password.toCharArray() : new char[0]);
            log.info("使用账号密码方式连接MQTT服务器，username={}", username);
        } else {
            log.info("使用匿名方式连接MQTT服务器");
        }

        // 清除会话 - 每次连接都作为新的会话
        options.setCleanSession(true);
        log.debug("MQTT连接配置: 启用清除会话");

        // 连接超时时间（毫秒）
        options.setConnectionTimeout(3000);
        log.debug("MQTT连接超时设置: {}ms", 3000);

        // 重连间隔时间（毫秒）
        options.setAutomaticReconnect(true);
        options.setMaxReconnectDelay(10000);
        log.debug("MQTT重连配置: 启用自动重连，最大重连延迟{}ms", 10000);

        factory.setConnectionOptions(options);
        log.info("MQTT客户端工厂初始化完成");
        return factory;
    }

    /** -----------------------发送信息----------------------*/

    /**
     * MQTT消息发送通道
     */
    @Bean
    public MessageChannel mqttOutboundChannel() {
        log.info("创建MQTT消息发送通道");
        return new DirectChannel();
    }

    /**
     * MQTT消息发送处理器
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        // 客户端ID添加时间戳确保唯一
        String clientId = clientIdPrefix + "sender-" + System.currentTimeMillis();
        log.info("创建MQTT消息发送处理器，客户端ID: {}", clientId);
        MqttPahoMessageHandler handler = new MqttPahoMessageHandler(
                clientId,
                mqttPahoClientFactory()
        );

        // 异步发送
        handler.setAsync(true);
        log.debug("MQTT消息发送模式: 异步");

        // 默认QoS级别
        handler.setDefaultQos(1);
        log.debug("MQTT默认QoS级别: 1");

        // 默认主题（可以在发送消息时动态指定）
        handler.setDefaultTopic("platform/commands");
        log.info("MQTT默认发送主题: platform/commands");
        return handler;
    }

    /** -----------------------接收信息----------------------*/

    /**
     * 接收消息
     */
    @Bean
    public MessageChannel mqttInputChannel(){
        log.info("创建MQTT消息接收通道");
        return new DirectChannel();
    }

    /**
     * MQTT消息接收适配器
     */
    @Bean
    public MessageProducer mqttInbound() {
        // 客户端ID添加时间戳确保唯一
        String clientId = clientIdPrefix + "receiver-" + System.currentTimeMillis();
        log.info("创建MQTT消息接收适配器，客户端ID: {}", clientId);

        // 创建消息驱动通道适配器
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                clientId,
                mqttPahoClientFactory(),
                subscribeTopic
        );

        // 消息转换
        adapter.setConverter(new DefaultPahoMessageConverter());

        // 设置QoS级别
        adapter.setQos(1);
        log.debug("MQTT接收QoS级别: 1");

        // 设置输出通道
        adapter.setOutputChannel(mqttInputChannel());
        log.info("MQTT消息接收适配器初始化完成，订阅主题: {}", subscribeTopic);
        return adapter;
    }

    /**
     * MQTT消息处理器 - 处理接收到的消息
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler mqttMessageHandler() {
        return message -> {
            try {
                // 获取消息主题
                String topic = message.getHeaders().get("mqtt_receivedTopic", String.class);

                // 获取消息内容
                String payload = message.getPayload().toString();

                log.info("收到MQTT消息 - 主题: {}, 内容: {}", topic, payload);

                // 这里可以添加消息处理逻辑
                // 例如：解析消息、验证格式、存储数据等

            } catch (Exception e) {
                log.error("处理MQTT消息失败", e);
            }
        };
    }

    /**
     * 当MQTT相关Bean初始化完成后打印启动成功日志
     */
    @Bean
    public String mqttStartupSuccessIndicator() {
        String message = "MQTT服务配置完成，Broker: " + mqttServer +
                (StringUtils.hasText(username) ? ", Auth: 用户名/密码" : ", Auth: 匿名");
        log.info("=============================================");
        log.info(message);
        log.info("订阅主题: {}", subscribeTopic);
        log.info("=============================================");
        return message;
    }
}